package com.weic.spark.scala.p2

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 * @Auther:BigData-weic
 * @ClassName:_02TransformationOps
 * @Date:2020/12/7 20:43
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
object _02TransformationOps {
	def main(args: Array[String]): Unit = {
		val conf = new SparkConf()
    		.setAppName("TransformationOps")
    		.setMaster("local[*]")
		val sc = new SparkContext(conf)

//		groupBykeyOps(sc)
//		reduceByKeyOps(sc)
//		foldByKeyOps(sc)

		sc.stop()
	}

	def foldByKeyOps(sc: SparkContext)={
		case class Student(id: Int, name:String, province: String)
		val stuRDD = sc.parallelize(List(
			Student(1, "卫超", "山西"),
			Student(2, "昊天", "福建"),
			Student(3, "苏江", "上海"),
			Student(4, "大志", "山西"),
			Student(5, "龙哥", "山西"),
			Student(6, "浩东", "北京"),
			Student(7, "忠忠", "北京")
		),2).mapPartitionsWithIndex((index,partition) => {
			val parList = partition.toList
			println(s"分区编号：${index}-->分区:${parList.mkString("[",",","]")}")
			parList.toIterator
		})
		//函数foldByKey
		val provinceRDD:RDD[(String,Int)] = stuRDD.map(stu => (stu.province,1)).foldByKey(0)(_+_)
		provinceRDD.foreach{case (province,count) => {
			println(s"城市：${province}--->人数${count}")
		}}

	}

	//reduceByKey
	def reduceByKeyOps(sc: SparkContext)={
		case class Student(id: Int, name:String, province: String)
		val stuRDD = sc.parallelize(List(
			Student(1, "卫超", "山西"),
			Student(2, "昊天", "福建"),
			Student(3, "苏江", "上海"),
			Student(4, "大志", "山西"),
			Student(5, "龙哥", "山西"),
			Student(6, "浩东", "北京"),
			Student(7, "忠忠", "北京")
		))
		val provinceRDD:RDD[(String,Int)] = stuRDD.map(stu => (stu.province,1)).reduceByKey(_+_)
		provinceRDD.foreach{case (province,count) => {
			println(s"城市：${province}--->人数${count}")
		}}

	}

	def groupBykeyOps(sc: SparkContext)={
		case class Student(id: Int, name:String, province: String)
		val stuRDD = sc.parallelize(List(
			Student(1, "卫超", "山西"),
			Student(2, "昊天", "福建"),
			Student(3, "苏江", "上海"),
			Student(4, "大志", "山西"),
			Student(5, "龙哥", "山西"),
			Student(6, "浩东", "北京"),
			Student(7, "忠忠", "北京")
		))
//		val provinceRDD:RDD[(String,Iterable[Student])] = stuRDD.groupBy(stu => stu.province)
		val provinceRDD:RDD[(String,Iterable[Student])] = stuRDD.map(stu => (stu.province,stu)).groupByKey()
		//dayin
		provinceRDD.foreach{case(province,stus) => {
			println(s"城市：${province}--->学生信息：${stus.mkString("[",",","]")}--->学生人数：${stus.size}")
		}}
	}
}
